package com.flinkTest;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.joda.time.DateTime;

import java.util.Iterator;

/**
 * Created by cola on 2022/5/3.
 */
public class FlinkDemo {

    public static void main(String[] args) throws Exception {
        // parse the parameters
        final ParameterTool params = ParameterTool.fromArgs(args);

        final long rate = params.getLong("rate", 300L);


        // obtain execution environment, run this example in "ingestion time"
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // make parameters available in the web interface
        env.getConfig().setGlobalJobParameters(params);

        // create the data sources for both grades and salaries
        DataStream<Tuple2<String, Integer>> grades = TestSource.getSource(env, rate);

        grades
                .keyBy(0)
                .timeWindow(Time.seconds(10))
                .aggregate(new AggregateFunction<Tuple2<String, Integer>, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(Tuple2<String, Integer> stringIntegerTuple2, Long aLong) {
                        return aLong+1;
                    }

                    @Override
                    public Long getResult(Long aLong) {
                        return aLong;
                    }

                    @Override
                    public Long merge(Long aLong, Long acc1) {
                        return aLong+acc1;
                    }
                },
                new WindowFunction<Long, String, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple k, TimeWindow w, Iterable<Long> itV, Collector<String> collector) throws Exception {
                        long ts = w.getStart();
                        String name = k.toString();
                        long n =0;
                        Iterator<Long> it = itV.iterator();
                        while(it.hasNext()){
                            n+=it.next();
                        }
                        collector.collect(new DateTime(ts)+"\t"+k.getField(0)+"\t"+n);
                    }
                })
                .setParallelism(1)
                .print()
                ;
        env.execute("demo");
    }
}
